package com.kafka.config;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.ContainerProperties;

/**
 * Kafka配置类 负责配置Kafka生产者和消费者的相关参数 从kafka.yml配置文件中读取配置项
 *
 * 主要配置项说明： - bootstrap-servers: Kafka集群地址(必填) - producer.acks:
 * 消息确认机制(all/-1/0/1, 默认all) - producer.retries: 发送失败重试次数(建议3-5) -
 * producer.batch-size: 批量发送大小(默认16384) - consumer.group-id: 消费者组ID(必填) -
 * consumer.auto-offset-reset: 偏移量重置策略(earliest/latest/none) -
 * consumer.max-poll-records: 单次拉取最大消息数(默认500)
 *
 * 性能调优建议： - 高吞吐场景可增大batch-size和buffer-memory - 低延迟场景可减小batch-size -
 * 严格顺序场景设置max.in.flight.requests.per.connection=1
 *
 * 异常处理： - 生产者重试可能导致消息重复 - 消费者需处理OffsetCommitException
 */
@Configuration
@PropertySource("classpath:kafka.yml")
public class KafkaConfig {

	// Kafka服务器地址
	@Value("${spring.kafka.bootstrap-servers}")
	private String bootstrapServers;

	// 生产者配置 - 确认机制
	@Value("${spring.kafka.producer.acks}")
	private String acks;

	// 生产者配置 - 重试次数
	@Value("${spring.kafka.producer.retries}")
	private int retries;

	// 生产者配置 - 批量发送大小
	@Value("${spring.kafka.producer.batch-size}")
	private int batchSize;

	// 生产者配置 - 缓冲区大小
	@Value("${spring.kafka.producer.buffer-memory}")
	private int bufferMemory;

	// 消费者配置 - 消费者组ID
	@Value("${spring.kafka.consumer.group-id}")
	private String groupId;

	// 消费者配置 - 偏移量重置策略
	@Value("${spring.kafka.consumer.auto-offset-reset}")
	private String autoOffsetReset;

	// 消费者配置 - 每次拉取最大记录数
	@Value("${spring.kafka.consumer.max-poll-records}")
	private int maxPollRecords;

	/**
	 * 配置Kafka生产者工厂
	 * 
	 * @return ProducerFactory 生产者工厂实例
	 * 
	 *         核心配置项： - bootstrap.servers: Kafka集群地址 - acks: 消息确认机制(all/-1/0/1) -
	 *         retries: 发送失败重试次数(建议配合retry.backoff.ms) - batch.size: 批量发送大小(字节) -
	 *         buffer.memory: 生产者缓冲区大小(字节)
	 * 
	 *         序列化配置： - key.serializer: 键序列化器(StringSerializer) - value.serializer:
	 *         值序列化器(StringSerializer)
	 * 
	 *         性能调优： - linger.ms: 批量发送等待时间(毫秒) -
	 *         max.in.flight.requests.per.connection: 最大在途请求数
	 * 
	 *         异常处理： - 重试可能导致消息重复，需业务端去重 - 缓冲区满会抛出TimeoutException
	 */
	@Bean
	public ProducerFactory<String, String> producerFactory() {
		Map<String, Object> configProps = new HashMap<>();
		configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
		configProps.put(ProducerConfig.ACKS_CONFIG, acks);
		configProps.put(ProducerConfig.RETRIES_CONFIG, retries);
		configProps.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
		configProps.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
		configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
				"org.apache.kafka.common.serialization.StringSerializer");
		configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
				"org.apache.kafka.common.serialization.StringSerializer");
		return new DefaultKafkaProducerFactory<>(configProps);
	}

	@Bean
	public KafkaTemplate<String, String> kafkaTemplate() {
		return new KafkaTemplate<>(producerFactory());
	}

	/**
	 * 配置Kafka消费者工厂
	 * 
	 * @return ConsumerFactory 消费者工厂实例
	 * 
	 *         核心配置项： - bootstrap.servers: Kafka集群地址 - group.id: 消费者组ID(必填) -
	 *         auto.offset.reset: 偏移量重置策略(earliest/latest/none) - max.poll.records:
	 *         单次拉取最大消息数
	 * 
	 *         反序列化配置： - key.deserializer: 键反序列化器(StringDeserializer) -
	 *         value.deserializer: 值反序列化器(StringDeserializer)
	 * 
	 *         性能调优： - fetch.min.bytes: 最小拉取字节数 - fetch.max.wait.ms: 最大等待时间 -
	 *         max.poll.interval.ms: 处理超时时间
	 * 
	 *         偏移量提交： - enable.auto.commit: 禁用自动提交(false) -
	 *         需配合@KafkaListener和Acknowledgment手动提交
	 * 
	 *         注意事项： - 长耗时处理需调整max.poll.interval.ms - 高并发需配置concurrency参数 -
	 *         需处理OffsetCommitException
	 */
	@Bean
	public ConsumerFactory<String, String> consumerFactory() {
		Map<String, Object> configProps = new HashMap<>();
		configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
		configProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
		configProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
		configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
				"org.apache.kafka.common.serialization.StringDeserializer");
		configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
				"org.apache.kafka.common.serialization.StringDeserializer");
		configProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
		configProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
		return new DefaultKafkaConsumerFactory<>(configProps);
	}

	/**
	 * 配置Kafka监听容器工厂
	 * 
	 * @return ConcurrentKafkaListenerContainerFactory 监听容器工厂实例
	 * 
	 *         核心配置： - consumerFactory: 消费者工厂实例 - concurrency: 并发消费者数量(默认1)
	 * 
	 *         消息确认模式： - MANUAL_IMMEDIATE: 手动立即提交 - MANUAL: 手动批量提交 - RECORD: 逐条自动提交
	 * 
	 *         批量处理： - setBatchListener(true): 启用批量监听 -
	 *         需配合ConsumerConfig.MAX_POLL_RECORDS_CONFIG
	 * 
	 *         错误处理： - setErrorHandler(): 自定义错误处理器 - setRetryTemplate(): 重试策略
	 * 
	 *         性能调优： - setConcurrency(): 根据分区数设置 - setPollTimeout(): 拉取超时时间
	 */
	@Bean
	public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
		ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
		factory.setConsumerFactory(consumerFactory());
		factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
		return factory;
	}
}